 1.Kafka高级特性解析之主题中的偏移量管理
   
   Kafka 1.0.2，__consumer_offsets主题中保存各个消费组的偏移量。
   早期由zookeeper管理消费组的偏移量。
   查询方法：
   通过原生 kafka 提供的工具脚本进行查询。
   工具脚本的位置与名称为bin/kafka-consumer-groups.sh
   首先运行脚本，查看帮助：
   参数                说明
   --all-topics        将所有关联到指定消费组的主题都划归到reset-offsets 操作范围
   --bootstrap-        必须：(基于消费组的新的消费者): 要连接的服务器地址。
   server <String:
  server to connect
   to>
   --by-duration      距离当前时间戳的一个时间段。格式：'PnDTnHnMnS'
  <String: duration>
  --command-          指定配置文件，该文件内容传递给Admin Client和消费者。
  config <String:     
  command config
  property file>
  --delete           传值消费组名称，删除整个消费组与所有主题的各个分区偏移量和所有者关系。
                     如： --group g1 --group g2 。
                     传值消费组名称和单个主题，仅删除该消费组到指定主题的分区偏移量和所属关系。
                     如： --group g1 --group g2 --topic t1 。
					 传值一个主题名称，仅删除指定主题与所有消费组分区偏移量以及所属关系。
                     如： --topic t1
					 注意：消费组的删除仅对基于ZK保存偏移量的消费组有效，
					 并且要小心使用，仅删除不活跃的消费组。
  --describe         描述给定消费组的偏移量差距（有多少消息还没有消费）。
  --execute          执行操作。支持的操作： reset-offsets 。
  --export           导出操作的结果到CSV文件。支持的操作： reset-offsets 。
  --from-file        重置偏移量到CSV文件中定义的值。
  <String: path to   
   CSV file>
  --group <String:   消费组
  consumer group>    
  --list             列出所有消费组。
  --new-consumer     使用新的消费者实现。这是默认值。随后的发行版中会删除这一操作。
  --reset-offsets     重置消费组的偏移量。当前一次操作只支持
                     一个消费组，并且该消费组应该是不活跃的
					 有三个操作选项
                     1.(默认)plan：要重置哪个偏移量。
                     2. execute：执行reset-offsets 操作。
                     3. process：配合--export 将操作结果导出到CSV格式。
					 可以使用如下选项：
					 --to-datetime、--by-period、--to-earliest、--to-latest
					 、--shift-by、--from-file、--to-current 。
					 必须选择一个选项使用。
                     要定义操作的范围，使用：
					 --all-topics、--topic 。
                     必须选择一个，除非使用--from-file 选项。
  --shift-by         重置偏移量n个。n可以是正值，也可以是负值。
  <Long: numberof-
   offsets>
  --timeout          对某些操作设置超时时间。
  <Long: timeout     如：对于描述指定消费组信息，指定毫秒值的最⼤等待时间，以获取正常数据
   (ms)>             (如刚创建的消费组，或者消费组做了一些更改操作)。默认时间：5000 。
  --to-current       重置到当前的偏移量。
  --to-datetime      重置偏移量到指定的时间戳。格式：'YYYY-MM-DDTHH:mm:SS.sss'
  <String: datetime>
  --to-earliest      重置为最早的偏移量
  --to-latest        重置为最新的偏移量
  --to-offset        重置到指定的偏移量。
  <Long: offset>
  --topic <String:   指定哪个主题的消费组需要删除，或者指定哪个主题的消费组需要包含到resetoffsets
  topic>             操作中。对于reset-offsets 操作，还可以指定分区： topic1:0,1,2 。其中
                     0,1,2 表示要包含到操作中的分区号。重置偏移量的操作支持多个主题一起操作。
  --zookeeper        必须，它的值，你懂的。--zookeeper linux121:2181/myKafka 。
  <String: urls>
  这里我们先编写一个生产者，消费者的例子：
  我们先启动消费者，再启动生产者， 再通过 bin/kafka-consumer-groups.sh 进⾏消费偏移量查询，
  由于kafka 消费者记录group的消费偏移量有两种方式 ：
  1)kafka 自维护 （新）
  2)zookpeer 维护 (旧) ，已经逐渐被废弃
  所以,脚本只查看由broker维护的，由zookeeper维护的可以将--bootstrap-server换成--zookeeper即可。
  1).查看有那些 group ID 正在进行消费：
  [root@linux121 ~]# kafka-consumer-groups.sh --bootstrap-server linux121:9092 --list
   Note: This will not show information about old Zookeeper-based consumers.
   group
   注意：
   (1).这里面是没有指定 topic，查看的是所有topic消费者的 group.id 的列表。
   (2).注意： 重名的 group.id 只会显示一次
  2).查看指定group.id 的消费者消费情况
  [root@linux121 ~]# kafka-consumer-groups.sh --bootstrap-server linux121:9092 --describe --
group group
  TOPIC      PARTITION   CURRENT-OFFSET   LOG-END-OFFSET   LAG
CONSUMER-ID                                      HOST
CLIENT-ID
  tp_demo_02   0             923              923           0
consumer-1-6d88cc72-1bf1-4ad7-8c6c-060d26dc1c49 /192.168.100.1
consumer-1
tp_demo_02     1             872              872           0
consumer-1-6d88cc72-1bf1-4ad7-8c6c-060d26dc1c49 /192.168.100.1
consumer-1
tp_demo_02     2             935              935           0
consumer-1-6d88cc72-1bf1-4ad7-8c6c-060d26dc1c49 /192.168.100.1
consumer-1
  如果消费者停止，查看偏移量信息：
  kafka-consumer-groups.sh --bootstrap-server linux121:9092 --describe --group group
  将偏移量设置为最早的：
  kafka-consumer-groups.sh --bootstrap-server linux121:9092 --reset-offsets --group group
--topic tp_demo_02 --to-earliest --execute 
  将偏移量设置为最新的：
  kafka-consumer-groups.sh --bootstrap-server linux121:9092 --reset-offsets --group group
--topic tp_demo_02 --to-latest 
  分别将指定主题的指定分区的偏移量向前移动10个消息：
  kafka-consumer-groups.sh --bootstrap-server linux121:9092 --reset-offsets --group group
--topic tp_demo_02:2 --shift-by -10 --execute
  kafka-consumer-groups.sh --bootstrap-server linux121:9092 --reset-offsets --group group
--topic tp_demo_02:1 --shift-by -10 --execute
  kafka-consumer-groups.sh --bootstrap-server linux121:9092 --reset-offsets --group group
--topic tp_demo_02:0 --shift-by -10 --execute
  kafka-consumer-groups.sh --bootstrap-server linux121:9092 --reset-offsets --group group
--topic tp_demo_02:0 --topic tp_demo_02:0,2 --shift-by -5 --execute
  代码：
  KafkaProducerSingleton.java
package com.lagou.kafka.demo.producer;

import org.apache.kafka.clients.producer.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.io.InputStream;
import java.util.Properties;
import java.util.Random;

public class KafkaProducerSingleton {
    private static final Logger LOGGER =
            LoggerFactory.getLogger(KafkaProducerSingleton.class);
    private static KafkaProducer<String, String> kafkaProducer;
    private Random random = new Random();
    private String topic;
    private int retry;

    private KafkaProducerSingleton() {

    }

    /**
     * 静态内部类
     *
     * @author tanjie
     */
    private static class LazyHandler {
        private static final KafkaProducerSingleton instance = new
                KafkaProducerSingleton();
    }

    /**
     * 单例模式,kafkaProducer是线程安全的,可以多线程共享一个实例
     *
     * @return
     */
    public static final KafkaProducerSingleton getInstance() {
        return LazyHandler.instance;
    }

    /**
     * kafka生产者进行初始化
     *
     * @return KafkaProducer
     */
    public void init(String topic, int retry) {
        this.topic = topic;
        this.retry = retry;
        if (null == kafkaProducer) {
            Properties props = new Properties();
            props.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "linux121:9092");
            props.setProperty(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
                    "org.apache.kafka.common.serialization.StringSerializer");
            props.setProperty(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
                    "org.apache.kafka.common.serialization.StringSerializer");
            props.setProperty(ProducerConfig.ACKS_CONFIG, "1");
            kafkaProducer = new KafkaProducer<String, String>(props);
        }
    }

    /**
     * 通过kafkaProducer发送消息
     *
     * @param message
     */
    public void sendKafkaMessage(final String message) {
        ProducerRecord<String, String> record = new ProducerRecord<String, String>(
                topic, random.nextInt(3), "", message);
        kafkaProducer.send(record, new Callback() {
            public void onCompletion(RecordMetadata recordMetadata,
                                     Exception exception) {
                if (null != exception) {
                    LOGGER.error("kafka发送消息失败:" + exception.getMessage(), exception);
                    retryKakfaMessage(message);
                }
            }
        });
    }

    /**
     * 当kafka消息发送失败后,重试
     *
     * @param retryMessage
     */
    private void retryKakfaMessage(final String retryMessage) {
        ProducerRecord<String, String> record = new ProducerRecord<String, String>(
                topic, random.nextInt(3), "", retryMessage);
        for (int i = 1; i <= retry; i++) {
            try {
                kafkaProducer.send(record);
                return;
            } catch (Exception e) {
                LOGGER.error("kafka发送消息失败:" + e.getMessage(), e);
                retryKakfaMessage(retryMessage);
            }
        }
    }

    /**
     * kafka实例销毁
     */
    public void close() {
        if (null != kafkaProducer) {
            kafkaProducer.close();
        }
    }

    public String getTopic() {
        return topic;
    }

    public void setTopic(String topic) {
        this.topic = topic;
    }

    public int getRetry() {
        return retry;
    }

    public void setRetry(int retry) {
        this.retry = retry;
    }
}
  ProducerHandler.java
package com.lagou.kafka.demo.producer;

public class ProducerHandler implements Runnable {
    private String message;

    public ProducerHandler(String message) {
        this.message = message;
    }

    @Override
    public void run() {
        KafkaProducerSingleton kafkaProducerSingleton =
                KafkaProducerSingleton.getInstance();
        kafkaProducerSingleton.init("tp_demo_02", 3);
        int i = 0;
        while (true) {
            try {
                System.out.println("当前线程:" + Thread.currentThread().getName()
                        + "\t获取的kafka实例:" + kafkaProducerSingleton);
                kafkaProducerSingleton.sendKafkaMessage("发送消息: " + message + " " + (++i));
                Thread.sleep(100);
            } catch (Exception e) {
            }
        }
    }
}
  MyProducer.java
package com.lagou.kafka.demo.producer;

public class MyProducer {
    public static void main(String[] args) {
        Thread thread = new Thread(new ProducerHandler("hello lagou "));
        thread.start();
    }
}
  KafkaConsumerAuto.java
package com.lagou.kafka.demo.consumer;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.Arrays;
import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.TimeUnit;

public class KafkaConsumerAuto {
    /**
     * kafka消费者不是线程安全的
     */
    private final KafkaConsumer<String, String> consumer;
    private ExecutorService executorService;

    public KafkaConsumerAuto() {
        Properties props = new Properties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "linux121:9092");
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "group");
        // 关闭自动提交
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
        props.put("auto.commit.interval.ms", "100");
        props.put(ConsumerConfig.SESSION_TIMEOUT_MS_CONFIG, "30000");
        props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringDeserializer");
        props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG,
                "org.apache.kafka.common.serialization.StringDeserializer");
        consumer = new KafkaConsumer<String, String>(props);
        // 订阅主题
        consumer.subscribe(Collections.singleton("tp_demo_02"));
    }

    public void execute() throws InterruptedException {
        executorService = Executors.newFixedThreadPool(2);
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(2_000);
            if (null != records) {
                executorService.submit(new ConsumerThreadAuto(records, consumer));
            }
            Thread.sleep(1000);
        }
    }

    public void shutdown() {
        try {
            if (consumer != null) {
                consumer.close();
            }
            if (executorService != null) {
                executorService.shutdown();
            }
            if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) {
                System.out.println("关闭线程池超时。。。");
            }
        } catch (InterruptedException ex) {
            Thread.currentThread().interrupt();
        }
    }
}
  ConsumerThreadAuto.java
package com.lagou.kafka.demo.consumer;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

public class ConsumerThreadAuto implements Runnable {
    private ConsumerRecords<String, String> records;
    private KafkaConsumer<String, String> consumer;

    public ConsumerThreadAuto(ConsumerRecords<String, String> records,
                              KafkaConsumer<String, String> consumer) {
        this.records = records;
        this.consumer = consumer;
    }

    @Override
    public void run() {
        for (ConsumerRecord<String, String> record : records) {
            System.out.println("当前线程:" + Thread.currentThread()
                    + "\t主题:" + record.topic()
                    + "\t偏移量:" + record.offset() + "\t分区:" + record.partition()
                    + "\t获取的消息:" + record.value());
        }
    }
}
  ConsumerAutoMain.java
package com.lagou.kafka.demo.consumer;

public class ConsumerAutoMain {
    public static void main(String[] args) {
        KafkaConsumerAuto kafka_consumerAuto = new KafkaConsumerAuto();
        try {
            kafka_consumerAuto.execute();
            Thread.sleep(20000);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } finally {
            kafka_consumerAuto.shutdown();
        }
    }
}
	
	  